package com.se.netty;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.handler.timeout.ReadTimeoutHandler;
import io.netty.util.CharsetUtil;
import lombok.extern.log4j.Log4j2;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.netty.DisposableServer;
import reactor.netty.tcp.TcpClient;
import reactor.netty.tcp.TcpServer;

import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.LockSupport;

/**
 * @Author: hdu_huang
 * @Date: 2018/12/28 10:19
 */
@Log4j2 //arthas
public class NettyReactor {
    public static void main(String[] args) {

        ServerHandler serverHandler = new ServerHandler();

        //ChannelOperationsHandler
        TcpServer server = TcpServer.create().host("127.0.0.1").port(9900)
//                .doOnBind(serverBootstrap -> log.error("doOnBind1"))
//                .doOnConnection(connection -> log.error("doOnConnection1"))
//                .doOnBound(disposableServer -> log.error("doOnBound1 : {}", disposableServer))
//                .doOnBind(serverBootstrap -> serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
//            @Override
//            protected void initChannel(SocketChannel socketChannel) throws Exception {
//
//                log.error("initChannel");
//                socketChannel.pipeline().
//                        addLast(new ReadTimeoutHandler(120)).  // 读超时（客户端长时间没有发送数据给服务端）
//                        addLast("logger", new LoggingHandler(LogLevel.DEBUG)).
//                        addLast(serverHandler); // 自定义 ChannelHandler
//            }
//        }))
//                .doOnBind(serverBootstrap -> {serverBootstrap.childHandler(serverHandler);log.error("doOnBind2");})
//                .doOnBound(disposableServer -> log.error("doOnBound2 : {}", disposableServer))

                .doOnConnection(connection -> connection.
//                        addHandler(new StringDecoder(CharsetUtil.UTF_8)).
//                        addHandler(new StringEncoder(CharsetUtil.UTF_8)).
//                        addHandler(new ReadTimeoutHandler(120)).
                        addHandler(new LoggingHandler(LogLevel.DEBUG))
                                .addHandler(serverHandler)
                )
//                .handle((inbound, outbound) -> {
//                    log.error("handle handle handle!!!");
//                    return inbound.receive().asString().flatMap(s -> {
//                        log.error("flatMap flatMap flatMap: {}", s);
//                        return outbound.sendString(Mono.just("hello"));
//                    });
//
//                })

                .handle((in, out) ->
                        in.receive()
                                .asString()
                                .flatMap(s -> {
                                    try {
//                                        Path file = Paths.get("");
//                                        return out.sendFile(file).then();
                                        log.error(s);

                                        return out.sendString(Mono.just("hello"));
                                    } catch (RuntimeException e) {
                                        return Mono.error(e).then();
                                    }
                                }))
                ;
//
        DisposableServer block = server.bind().block();
        Channel channel = block.channel();
//
        log.info(channel.id());


//        createTcpClient();

        LockSupport.park();
    }

    private static void createTcpClient() {


        TcpClient.create()
//                .doOnConnect(bootstrap -> bootstrap.connect("localhost", 9999))
                .doOnConnect(bootstrap -> log.error("doOnConnect"))
                .doOnConnected(connection -> connection.addHandler(new StringDecoder(CharsetUtil.UTF_8)).
                        addHandler(new StringEncoder(CharsetUtil.UTF_8)).
                        addHandler("ping", new IdleStateHandler(10, 0, 0, TimeUnit.SECONDS)).
                        addHandler(new ClientHandler()))
                .doOnDisconnected(connection -> log.error("doOnDisconnected"))
                .host("localhost").port(9999).connect().block();

    }
}
